package com.atguigu.chapter11;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/6/19 9:26
 */
public class Flink05_SQL_BaseUse_1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 1. 创建一个流
        DataStreamSource<WaterSensor> stream = env.fromElements(
            new WaterSensor("sensor_1", 1000L, 10),
            new WaterSensor("sensor_1", 2000L, 20),
            new WaterSensor("sensor_2", 3000L, 30),
            new WaterSensor("sensor_1", 4000L, 40)
        );
        // 2. 创建流式表的环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        // 3. 把流转成动态表
        Table table = tEnv.fromDataStream(stream);
        
        // 4. 在table上使用sql语句进行查询
        
        //tEnv.executeSql()   // 执行ddl 增删改
        
        // tEnv.sqlQuery()  // 执行查询
        
        // 从一个表中查询数据:   未注册的表
        /*Table result = tEnv.sqlQuery("select " +
                                         "  * " +
                                         " from " + table +
                                         " where id='sensor_1'");*/
        
        // 把table注册一个临时表
        tEnv.createTemporaryView("sensor", table);
        Table result = tEnv.sqlQuery("select " +
                                         "  * " +
                                         " from sensor " +
                                         " where id='sensor_1'");
    
        result.execute().print();
    }
}
